package com.avoid.easymqtt.remoting.netty;

import com.avoid.easymqtt.common.SystemClock;
import com.avoid.easymqtt.mqtt.AsyncMqttRequestProcessor;
import com.avoid.easymqtt.mqtt.MqttRequestProcessor;
import com.avoid.easymqtt.mqtt.MqttResponseCallback;
import com.avoid.easymqtt.mqtt.protocol.MqttMessageFactory;
import com.avoid.easymqtt.remoting.ChannelEventListener;
import com.avoid.easymqtt.remoting.common.Pair;
import com.avoid.easymqtt.remoting.common.RemotingHelper;
import com.avoid.easymqtt.remoting.common.ServiceThread;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.mqtt.MqttConnAckMessage;
import io.netty.handler.codec.mqtt.MqttConnectReturnCode;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttProperties;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;

/**
 * 描述
 *
 * @author avoid@fiture.com
 * @date 2022/4/29
 * @motto Life is so short,do something to make yourself happy,such as coding
 */
public abstract class NettyRemotingAbstract {

    private static final Logger log = LoggerFactory.getLogger(RemotingHelper.EASYMQTT_REMOTING);

    /**
     * 这个容器包含每个请求代码（aka）的所有处理器。对于每个传入的请求，我们可以在这个映射中查找响应的处理器来处理请求。
     */
    protected final HashMap<Integer, Pair<MqttRequestProcessor, ExecutorService>> processorTable = new HashMap<>(64);

    /**
     * 在每个请求代码的{@link #processorTable}中没有精确匹配时使用的默认请求处理器。
     */
    protected Pair<MqttRequestProcessor, ExecutorService> defaultRequestProcessor;

    /**
     * 执行器将netty事件提供给用户定义的 {@link ChannelEventListener}.
     */
    protected final NettyEventExecutor nettyEventExecutor = new NettyEventExecutor();

    /**
     * 要创建的SSL上下文 {@link SslHandler}.
     */
    protected volatile SslContext sslContext;

    /**
     * 自定义通道事件侦听器。
     *
     * @return 自定义通道事件侦听器（如果已定义）；否则为空。
     */
    public abstract ChannelEventListener getChannelEventListener();

    /**
     * 将netty事件交由执行器执行
     *
     * @param event Netty event instance.
     */
    public void putNettyEvent(final NettyEvent event) {
        this.nettyEventExecutor.putNettyEvent(event);
    }

    /**
     * 传输信息处理
     *
     * <p>
     * <strong>Note:</strong> 传入的远程处理命令可能是
     * <ul>
     * <li>来自远程对等组件的查询请求；</li>
     * <li>对该参与者之前发出的请求的回应。</li>
     * </ul>
     * </p>
     *
     * @param ctx 通道处理程序上下文。
     * @param request 传入MQTT信息。
     * @throws Exception 如果在处理传入命令时出现任何错误。
     */
    public void processMessageReceived(ChannelHandlerContext ctx, MqttMessage request) throws Exception {
        int mqttMessageType = request.fixedHeader().messageType().value();
        final Pair<MqttRequestProcessor, ExecutorService> matched = this.processorTable.get(mqttMessageType);
        final Pair<MqttRequestProcessor, ExecutorService> pair =
            null == matched ? this.defaultRequestProcessor : matched;
        if (pair != null) {
            Runnable run = () -> {
                try {
                    MqttRequestProcessor processor = pair.getObject1();
                    final MqttResponseCallback callback = new MqttResponseCallback() {
                        @Override
                        public void callback(MqttMessage response) {
                            if (response != null) {
                                try {
                                    ctx.writeAndFlush(response);
                                } catch (Throwable e) {
                                    log.error("处理请求已结束，但响应失败", e);
                                    log.error(request.toString());
                                    log.error(response.toString());
                                }
                            }
                        }
                    };
                    if (processor instanceof AsyncMqttRequestProcessor) {
                        AsyncMqttRequestProcessor asyncProcessor = (AsyncMqttRequestProcessor)processor;
                        asyncProcessor.asyncProcessRequest(ctx, request, callback);
                    } else {
                        MqttMessage response = processor.processRequest(ctx, request);
                        callback.callback(response);
                    }
                } catch (Throwable e) {
                    log.error("process request exception", e);
                    log.error(request.toString());

                    String error = "[REJECT_REQUEST]system busy, start flow control for a while";
                    MqttProperties mqttProperties = new MqttProperties();
                    mqttProperties.add(new MqttProperties.StringProperty(
                        MqttProperties.MqttPropertyType.REASON_STRING.value(), error));
                    MqttConnAckMessage connectAckMessage = MqttMessageFactory.getConnectAckMessage(
                        MqttConnectReturnCode.CONNECTION_REFUSED_SERVER_UNAVAILABLE, false, mqttProperties);
                    ctx.writeAndFlush(connectAckMessage);
                }
            };
            // 是否被拒绝
            if (pair.getObject1().rejectRequest()) {
                String error = "[REJECT_REQUEST]system busy, start flow control for a while";
                MqttProperties mqttProperties = new MqttProperties();
                mqttProperties.add(
                    new MqttProperties.StringProperty(MqttProperties.MqttPropertyType.REASON_STRING.value(), error));
                MqttConnAckMessage connectAckMessage = MqttMessageFactory.getConnectAckMessage(
                    MqttConnectReturnCode.CONNECTION_REFUSED_SERVER_UNAVAILABLE, false, mqttProperties);
                ctx.writeAndFlush(connectAckMessage);
            }

            try {
                final RequestTask requestTask = new RequestTask(run, ctx.channel(), request);
                pair.getObject2().submit(requestTask);
            } catch (RejectedExecutionException e) {
                if ((SystemClock.getNow() % 10000) == 0) {
                    log.warn(RemotingHelper.parseChannelRemoteAddr(ctx.channel())
                        + ", 请求太多，系统线程池忙, RejectedExecutionException " + pair.getObject2().toString()
                        + " mqtt message type: " + mqttMessageType);
                }

                MqttProperties mqttProperties = new MqttProperties();
                mqttProperties
                    .add(new MqttProperties.StringProperty(MqttProperties.MqttPropertyType.REASON_STRING.value(),
                        "[OVERLOAD]system busy, start flow control for a while"));
                MqttConnAckMessage connectAckMessage = MqttMessageFactory.getConnectAckMessage(
                    MqttConnectReturnCode.CONNECTION_REFUSED_SERVER_UNAVAILABLE, false, mqttProperties);
                ctx.writeAndFlush(connectAckMessage);
            }
        } else {
            String error = " mqtt message type " + mqttMessageType + " not supported";
            MqttProperties mqttProperties = new MqttProperties();
            mqttProperties
                .add(new MqttProperties.StringProperty(MqttProperties.MqttPropertyType.REASON_STRING.value(), error));
            MqttConnAckMessage connectAckMessage = MqttMessageFactory.getConnectAckMessage(
                MqttConnectReturnCode.CONNECTION_REFUSED_SERVER_UNAVAILABLE, false, mqttProperties);
            ctx.writeAndFlush(connectAckMessage);
            log.error(RemotingHelper.parseChannelRemoteAddr(ctx.channel()) + error);
        }
    }

    /**
     * <p>
     * 定期调用此方法以扫描不推荐的请求并使其过期。
     * </p>
     */
    public void scanResponseTable() {

    }

    class NettyEventExecutor extends ServiceThread {
        private final LinkedBlockingQueue<NettyEvent> eventQueue = new LinkedBlockingQueue<NettyEvent>();
        private final int maxSize = 10000;

        public void putNettyEvent(final NettyEvent event) {
            if (this.eventQueue.size() <= maxSize) {
                this.eventQueue.add(event);
            } else {
                log.warn("事件队列大小[{}] 不足, 所以，丢弃这个事件 {}", this.eventQueue.size(), event.toString());
            }
        }

        @Override
        public void run() {
            log.info(this.getServiceName() + " 服务启动");

            final ChannelEventListener listener = NettyRemotingAbstract.this.getChannelEventListener();

            while (!this.isStopped()) {
                try {
                    NettyEvent event = this.eventQueue.poll(3000, TimeUnit.MILLISECONDS);
                    if (event != null && listener != null) {
                        switch (event.getType()) {
                            case IDLE:
                                listener.onChannelIdle(event.getRemoteAddr(), event.getChannel());
                                break;
                            case CLOSE:
                                listener.onChannelClose(event.getRemoteAddr(), event.getChannel());
                                break;
                            case CONNECT:
                                listener.onChannelConnect(event.getRemoteAddr(), event.getChannel());
                                break;
                            case EXCEPTION:
                                listener.onChannelException(event.getRemoteAddr(), event.getChannel());
                                break;
                            default:
                                break;
                        }
                    }
                } catch (Exception e) {
                    log.warn(this.getServiceName() + " service has exception. ", e);
                }
            }
            log.info(this.getServiceName() + " service end");
        }

        @Override
        public String getServiceName() {
            return NettyEventExecutor.class.getSimpleName();
        }
    }
}
